package com.atguigu.flink.sql.window;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Expressions;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * Created by Smexy on 2023/4/11
 */
public class Demo1_DefineTimeStream
{
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(env);

        env.setParallelism(1);

        WatermarkStrategy<WaterSensor> watermarkStrategy = WatermarkStrategy
            .<WaterSensor>forMonotonousTimestamps()
            .withTimestampAssigner( (e, ts) -> e.getTs());


        //自带水印，自带eventtime
        SingleOutputStreamOperator<WaterSensor> ds = env
            .socketTextStream("hadoop102", 8888)
            .map(new WaterSensorMapFunction())
            //产生了水印，向下游发送水印
            .assignTimestampsAndWatermarks(watermarkStrategy);

        Schema schema = Schema.newBuilder()
                             //声明普通列
                             .column("id", "STRING")
                             .column("ts", "BIGINT")
                             .column("vc", "INT")
                             //这个列由一个表达式计算得到
                             .columnByExpression("pt", "proctime()")
                             .columnByExpression("et", "TO_TIMESTAMP_LTZ(ts,3)")
                             .build();
        //从流中获取时间属性
        Table table = tableEnvironment.fromDataStream(ds,schema);

        table.execute().print();

        env.execute();


    }
}
